package com.atguigu.app.dwd.db;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.utils.KafkaUtil;
import com.atguigu.utils.MySqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

//支付成功明细表 interval join要求两个表创建表时都要指定事件时间和watermark
//从下单到支付成功可以有15min中的时间间隔，所以下单表要设置状态时间为15min+5s(乱序程度)
//todo 1.获取环境
//todo 设置状态后端
//todo 2.创建动态表dwd_order_detail连接kafka dwd_order_detail主题
//todo 3.创建动态表topic_db连接kafka topic_db主题
//todo 4.过滤出topic_db中的支付表以及支付成功的数据
//todo 5.创建动态表base_dic连接mysql读取base_dic表
//todo 6.三表关联
//todo 7.创建动态表dwd_trade_pay_detail_suc连接kafka dwd_trade_pay_detail_suc表
//todo 8.将关联后的数据写到dwd_trade_pay_detail_suc中
public class DwdTradePayDetailSuc {
    public static void main(String[] args) {
        //todo 1.获取环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //todo 设置状态后端

        //todo 2.创建动态表dwd_order_detail连接kafka dwd_order_detail主题
        //此操作将null值过滤掉了，就剩左null和左右，由于本案例没有left join，所以用kafka和upsert kafka都可以，但是我们想让同一个订单明细id的左null和左右都进到一个分区，所以用upsert kafka，设置主键为id，相同id的key就会进到同一个分区
        tableEnv.executeSql("create table dwd_order_detail(\n" +
                "id string,\n" +
                "order_id string,\n" +
                "sku_id string,\n" +
                "sku_name string,\n" +
                "order_price string,\n" +
                "sku_num string,\n" +
                "create_time string,\n" +
                "source_type string,\n" +
                "source_id string,\n" +
                "split_total_amount string,\n" +
                "split_activity_amount string,\n" +
                "split_coupon_amount string,\n" +
                "total_amount string,\n" +
                "order_status string,\n" +
                "user_id string,\n" +
                "payment_way string,\n" +
                "order_create_time string,\n" +
                "order_operate_time string,\n" +
                "order_expire_time string,\n" +
                "process_status string,\n" +
                "province_id string,\n" +
                "activity_reduce_amount string,\n" +
                "coupon_reduce_amount string,\n" +
                "original_total_amount string,\n" +
                "feight_fee string,\n" +
                "feight_fee_reduce string,\n" +
                "refundable_time string,\n" +
                "activity_id string,\n" +
                "activity_rule_id string,\n" +
                "order_price string,\n" +
                "create_time string,\n" +
                "source_type string,\n" +
                "source_id string,\n" +
                "split_total_amount string,\n" +
                "split_activity_amount string,\n" +
                "split_coupon_amount string,\n" +
                "source_type_name string,\n" +
                "pt as proctime(),\n" +//lookup join
//                "rt as TO_TIMESTAMP(create_time),\n" +//interval join*********************
                "ts string," +
                "rt as TO_TIMESTAMP_LTZ(cast(ts as bigint)*1000,3)," +//maxwell是s级，我们要ms级
//                下面还有另外三种写法
//                "rt as TO_TIMESTAMP_LTZ(cast(ts as bigint),0)," +
                //"row_time as TO_TIMESTAMP(FROM_UNIXTIME(cast(ts as bigint))),\n" +
                //"row_time as TO_TIMESTAMP((cast(ts as bigint)*1000)),\n" +
                "watermark for rt as rt \n" +//不需要设置乱序程度，因为interval join会设置事件时间范围
                ")\n"+KafkaUtil.getKafkaConnectorDDL("dwd_trade_order_detail","DwdTradePayDetailSuc_220828"));



        //todo 3.创建动态表topic_db连接kafka topic_db主题
        tableEnv.executeSql(KafkaUtil.getKafkaTopicDbDDL("DwdTradePayDetailSuc_220828"));

        //todo 4.过滤出topic_db中的支付表以及支付成功的数据
        Table payDetailSucTable = tableEnv.sqlQuery("" +
                "select \n" +
                "data['id'] id,\n" +
                "data['out_trade_no'] out_trade_no,\n" +
                "data['order_id'] order_id,\n" +
                "data['user_id'] user_id,\n" +
                "data['payment_type'] payment_type,\n" +
                "data['trade_no'] trade_no,\n" +
                "data['total_amount'] total_amount,\n" +
                "data['subject'] subject,\n" +
                "data['payment_status'] payment_status,\n" +
                "data['create_time'] create_time,\n" +
                "data['callback_time'] callback_time,\n" +
                "rt " +//rt一定要取出来，后面要用
                "from topic_db\n" +
                "where `database`='gmall'\n" +
                "and `table`='payment_info'\n" +
                "and type='update'\n" +
                "and `old`['payment_status'] is not null\n" +
                "and data['payment_status']='1602'");
//        payDetailSucTable.execute().print();
        tableEnv.createTemporaryView("pay_suc",payDetailSucTable);

        //todo 5.创建动态表base_dic连接mysql读取base_dic表
        tableEnv.executeSql(MySqlUtil.getBaseDicDDL());

        //todo 6.三表关联
        Table intervalJoinTable = tableEnv.sqlQuery("select  \n" +
                "od.id,\n" +
                "od.order_id,\n" +
                "od.sku_id,\n" +
                "od.sku_name,\n" +
                "od.sku_num,\n" +
                "od.split_total_amount,\n" +
                "od.split_activity_amount,\n" +
                "od.split_coupon_amount,\n" +
                "od.total_amount,\n" +
                "od.order_status,\n" +
                "od.user_id,\n" +
                "od.payment_way,\n" +
                "od.province_id,\n" +
                "od.activity_reduce_amount,\n" +
                "od.coupon_reduce_amount,\n" +
                "od.original_total_amount,\n" +
                "od.feight_fee,\n" +
                "od.feight_fee_reduce,\n" +
                "od.refundable_time,\n" +
                "od.activity_id,\n" +
                "od.activity_rule_id,\n" +
                "od.order_price,\n" +
                "od.create_time,\n" +
                "od.source_type,\n" +
                "od.source_id,\n" +
                "od.split_total_amount,\n" +
                "od.split_activity_amount,\n" +
                "od.split_coupon_amount,\n" +
                "od.source_type_name,\n" +
                "od.pt," +

                "suc.payment_type,\n" +
                "suc.trade_no,\n" +
                "suc.total_amount,\n" +
                "suc.callback_time\n" +
                "from dwd_order_detail od,pay_suc suc \n" +
                "where od.order_id=suc.order_id\n" +
                "and suc.rt<=od.rt+interval '15' minute\n" +
                "and suc.rt>=od.rt-interval '5' second");
        tableEnv.createTemporaryView("interval_join_table",intervalJoinTable);

        Table resultTable = tableEnv.sqlQuery("select i.order_id,\n" +
                "i.sku_id,\n" +
                "i.sku_name,\n" +
                "i.sku_num,\n" +
                "i.split_total_amount,\n" +
                "i.split_activity_amount,\n" +
                "i.split_coupon_amount,\n" +
                "i.total_amount,\n" +
                "i.order_status,\n" +
                "i.user_id,\n" +
                "i.payment_way,\n" +
                "i.province_id,\n" +
                "i.activity_reduce_amount,\n" +
                "i.coupon_reduce_amount,\n" +
                "i.original_total_amount,\n" +
                "i.feight_fee,\n" +
                "i.feight_fee_reduce,\n" +
                "i.refundable_time,\n" +
                "i.activity_id,\n" +
                "i.activity_rule_id,\n" +
                "i.order_price,\n" +
                "i.create_time,\n" +
                "i.source_type,\n" +
                "i.source_id,\n" +
                "i.payment_type,\n" +
                "i.trade_no,\n" +
                "i.total_amount,\n" +
                "i.callback_time," +
                "dic.dic_name from interval_join_table i " +
                "join base_dic for system_time as of i.pt as dic on i.source_type=dic.dic_code");
        tableEnv.createTemporaryView("result_table",resultTable);



        //todo 7.创建动态表dwd_trade_pay_detail_suc连接kafka dwd_trade_pay_detail_suc表
        tableEnv.executeSql("create table dwd_pay_detail(\n" +
                "id string,\n" +
                "order_id string,\n" +
                "sku_id string,\n" +
                "sku_name string,\n" +
                "sku_num string,\n" +
                "split_total_amount string,\n" +
                "split_activity_amount string,\n" +
                "split_coupon_amount string,\n" +
                "total_amount string,\n" +
                "order_status string,\n" +
                "user_id string,\n" +
                "payment_way string,\n" +
                "province_id string,\n" +
                "activity_reduce_amount string,\n" +
                "coupon_reduce_amount string,\n" +
                "original_total_amount string,\n" +
                "feight_fee string,\n" +
                "feight_fee_reduce string,\n" +
                "refundable_time string,\n" +
                "activity_id string,\n" +
                "activity_rule_id string,\n" +
                "order_price string,\n" +
                "create_time string,\n" +
                "source_type string,\n" +
                "source_id string,\n" +
                "payment_type string,\n" +
                "trade_no string,\n" +
                "total_amount string,\n" +
                "callback_time string," +
                "dic_name string," +
                "primary key (id) not enforced" +
                ")"+KafkaUtil.getUpsertKafkaDDL("dwd_trade_pay_detail_suc"));

//        //todo 8.将关联后的数据写到dwd_trade_pay_detail_suc中
        tableEnv.executeSql("insert into dwd_pay_detail select * from result_table");


    }
}
